package com.yootk.rockemq;

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

import java.util.List;

public class MessagePullTagConsumer { // 消息消费者
    public static final String NAMESPACE = "MuyanYootk"; // 定义命名空间
    // 如果此时你使用的是集群服务，则每个主机之间使用“,”分割
    public static final String NAME_SERVER_LIST = "rocketmq-server:9876"; // NameServer
    public static final String CONSUMER_GROUP = "muyan-groupx-tag"; // 定义消费组
    public static final String TOPIC = "TopicYootkNS"; // 定义主题名称
    public static final String ACCESS_KEY = "RocketMQMuyan"; // 定义用户名
    public static final String SECRET_KEY = "helloyootk"; // 定义密码
    public static void main(String[] args) throws Exception { // 懒人必备的处理形式
        // 1、由于此时的RocketMQ启动了ACL安全认证的保护机制，所以需要配置相应的回调
        RPCHook clientHook = new AclClientRPCHook(new SessionCredentials(ACCESS_KEY, SECRET_KEY));
        // 2、要在项目之中启动一个消费者的处理程序类，但是这个消费者分为两种形式
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(CONSUMER_GROUP, clientHook); // 使用PULL模式
        consumer.setNamesrvAddr(NAME_SERVER_LIST); // 通过NameServer获取相应的Broker数据
        consumer.setNamespace(NAMESPACE); // 配置命名空间
        consumer.subscribe(TOPIC, "emp||dept"); // 匹配两个标签
        consumer.start(); // 启动消息的消费者
        boolean running = true; // 接收的状态
        while (running) {   // 处理了一个死循环的概念
            List<MessageExt> messages = consumer.poll(); // 手工的进行消息的拉取
            if (messages != null) { // 因为可能拉取到消息，也可能没有拉取到
                for (Message msg : messages) {
                    System.out.printf("【%s】keys = %s、namespace = %s、body = %s %n", msg.getTags(), msg.getKeys(), consumer.getNamespace(), new String(msg.getBody()));
                }
            }
        }
        consumer.shutdown();
    }
}
